package org.lds.simple.disruptor.consumer;

import lombok.AllArgsConstructor;
import lombok.Data;
import org.lds.simple.disruptor.LifeCycle;
import org.lds.simple.disruptor.RingBuffer;
import org.lds.simple.disruptor.consumer.exception.ConsumeExceptionHandler;
import org.lds.simple.disruptor.event.Event;
import org.lds.simple.disruptor.producer.sequencer.Sequencer;
import org.lds.simple.disruptor.sequence.Sequence;
import org.lds.simple.disruptor.strategy.WaitStrategy;

@Data
@AllArgsConstructor
public class EventConsumerTask<T extends Event> implements Runnable, LifeCycle {
    private final Sequencer sequencer;
    private final RingBuffer<T> ringBuffer;
    private final EventConsumer eventConsumer;
    private final WaitStrategy waitStrategy;
    private final ConsumeExceptionHandler consumeExceptionHandler;

    private final int bufferMask;

    /**
     * 消费者序列号
     */
    private final Sequence sequence = new Sequence();
    private boolean running = true;

    public EventConsumerTask(Sequencer sequencer, RingBuffer ringBuffer, EventConsumer eventConsumer, WaitStrategy waitStrategy) {
        this(sequencer, ringBuffer, eventConsumer, waitStrategy, null);
    }

    public EventConsumerTask(Sequencer sequencer, RingBuffer ringBuffer, EventConsumer eventConsumer, WaitStrategy waitStrategy, ConsumeExceptionHandler consumeExceptionHandler) {
        this.sequencer = sequencer;
        this.ringBuffer = ringBuffer;
        this.bufferMask = ringBuffer.getBufferSize() - 1;
        this.eventConsumer = eventConsumer;
        this.waitStrategy = waitStrategy;
        this.consumeExceptionHandler = consumeExceptionHandler;
    }

    @Override
    public void run() {
        while (running) {
            long maxConsumerSequence = waitStrategy.consumerWait(sequencer.getSequence(), sequence);

            for (long currConsumerSequence = sequence.get() + 1; currConsumerSequence <= maxConsumerSequence; currConsumerSequence++) {
                T event = ringBuffer.getEvent(currConsumerSequence);

                try {
                    eventConsumer.consumer(event, currConsumerSequence);
                } catch (Throwable e) {
                    consumeExceptionHandler.handleException(event, currConsumerSequence, e);
                }
            }

            sequence.set(maxConsumerSequence);
        }
    }

    @Override
    public void start() {
    }

    @Override
    public void shutdown() {
        this.running = false;
    }
}
